package com.lzcu.maoyan.mapreduce.mapreduce;

import com.lzcu.maoyan.mapreduce.mapreduce.mapper.MovieTypeMapper;
import com.lzcu.maoyan.mapreduce.mapreduce.reduce.MovieTypeReduce;
import com.lzcu.maoyan.mapreduce.mapreduce.unit.MovieTypeWritable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.phoenix.mapreduce.PhoenixInputFormat;
import org.apache.phoenix.mapreduce.PhoenixOutputFormat;
import org.apache.phoenix.mapreduce.util.PhoenixMapReduceUtil;

/**
 * 电影类型分析
 * @author WangLiang
 * @date `2024/4/24`
 */
public class MovieTypeMR extends Configured implements Tool {

    @Override
    public int run(String[] strings) throws Exception {
        //1.创建job
        Configuration conf = new Configuration();
        conf.addResource("phoenix/core-site.xml");
        conf.addResource("phoenix/hbase-site.xml");
        conf.addResource("phoenix/hdfs-site.xml");
        Job job = Job.getInstance(conf, "PhoenixStockApplication");
        job.setJarByClass(PhoenixMR.class);
        //2.设置输入输出格式
        job.setInputFormatClass(PhoenixInputFormat.class);
        job.setOutputFormatClass(PhoenixOutputFormat.class);
        //3.设置数据读入和写出路径
        String selectQuery = "SELECT DISTINCT T2.TYPEMOVIE AS SOURCE_MOVIE_TYPE, T2.MOVIENAME AS SOURCE_MOVIE_NAME\n" +
                "FROM MAOYANMOVIE.movie_info T1 \n" +
                "INNER JOIN MAOYANMOVIE.movie_detailed T2 \n" +
                "ON (T1.MOVIE_NAME_INFO=T2.MOVIENAME AND YEAR(TO_DATE(T2.RELEASEDATE)) IN (2024))";
        PhoenixMapReduceUtil.setInput(job, MovieTypeWritable.class, "MAOYANMOVIE.movie_detailed", selectQuery);
        PhoenixMapReduceUtil.setOutput(job, "MAOYANMOVIE.movie_type", "MOVIE_TYPE,COUNT,MOVIE,MOVIE_TOTAL");
        //4.设置代码片段
        job.setMapperClass(MovieTypeMapper.class);
        job.setReducerClass(MovieTypeReduce.class);
        //5.设置Mapper和Reducer端的输出key,value类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(MovieTypeWritable.class);
        TableMapReduceUtil.addDependencyJars(job);
        //6.任务提交
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new MovieTypeMR(), args);
    }

}

/*
 * CREATE TABLE MAOYANMOVIE.movie_type (
 *     movie_Type VARCHAR NOT NULL,
 *     count BIGINT,
 *     movie VARCHAR,
 *     MOVIE_TOTAL BIGINT,
 *     CONSTRAINT pk PRIMARY KEY (movie_Type)
 * );
 */